/**
 * DataTableCursor.java
 * Copyright (C) 2011 Yuanyan<yuanyan.cao@gmail.com>
 * 
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; version 2 of the License.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * For information on how to redistribute this software under
 * the terms of a license other than GNU General Public License
 * contact yuanyan.cao@gmail.com
 */
package edu.rabbit.kernel.table;

import java.util.Collections;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;


import edu.rabbit.Database;
import edu.rabbit.DbErrorCode;
import edu.rabbit.DbException;
import edu.rabbit.kernel.DbUtility;
import edu.rabbit.kernel.btree.IBtree;
import edu.rabbit.schema.IIndexDef;
import edu.rabbit.schema.ITableDef;
import edu.rabbit.schema.ConflictAction;
import edu.rabbit.table.ICursor;
import edu.rabbit.table.RunnableWithLock;
import edu.rabbit.table.ITable;
import edu.rabbit.table.ITransaction;

/**
 * Implementation of {@link ITable}.
 * 
 * @author Yuanyan<yanyan.cao@gmail.com>
 * 
 * 
 */
public class Table implements ITable {

    private interface ITableRun {
        public Object run(final IBtreeDataTable table) throws DbException;
    }

    private final Database db;
    private IBtree btree;
    private String tableName;
    private boolean write;

    public Table(Database db, IBtree btree, String tableName, boolean write) throws DbException {
        this.db = db;
        this.btree = btree;
        this.tableName = tableName;
        this.write = write;
        if (null == getDefinition())
            throw new DbException(DbErrorCode.ERROR, "Table not found: " + tableName);
    }

    /*
     * (non-Javadoc)
     * 
     * @see org.tmatesoft.rabbit.core.table.ITable#getDataBase()
     */
    public Database getDataBase() {
        return db;
    }

    public String getPrimaryKeyIndexName() throws DbException {
        final ITableDef definition = getDefinition();
        return definition.isRowIdPrimaryKey() ? null : definition.getPrimaryKeyIndexName();
    }

    public ITableDef getDefinition() throws DbException {
        return btree.getSchema().getTable(tableName);
    };

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.tmatesoft.rabbit.core.table.ITable#getIndexes(java.lang.String)
     */
    public Set<IIndexDef> getIndexesDefs() throws DbException {
        return btree.getSchema().getIndexes(tableName);
    }

    /*
     * (non-Javadoc)
     * 
     * @see org.tmatesoft.rabbit.core.table.ITable#getIndexNames()
     */
    public Set<String> getIndexesNames() throws DbException {
        final Set<String> result = new TreeSet<String>(String.CASE_INSENSITIVE_ORDER);
        final Set<IIndexDef> indexesDefs = getIndexesDefs();
        if (null != indexesDefs) {
            for (final IIndexDef indexDef : indexesDefs) {
                if (null != indexDef) {
                    result.add(indexDef.getName());
                }
            }
        }
        return Collections.unmodifiableSet(result);
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.tmatesoft.rabbit.core.table.ITable#getIndex(java.lang.String)
     */
    public IIndexDef getIndexDef(String name) throws DbException {
        if (null == name) {
            name = getPrimaryKeyIndexName();
            if (null == name) {
                return null;
            }
        }
        final Set<IIndexDef> indexesDefs = getIndexesDefs();
        if (null != indexesDefs) {
            for (final IIndexDef indexDef : indexesDefs) {
                if (null != indexDef && name.equalsIgnoreCase(indexDef.getName())) {
                    return indexDef;
                }
            }
        }
        return null;
    }

    public ICursor open() throws DbException {
        return (ICursor) db.runWithLock(new RunnableWithLock() {
            public Object runWithLock(Database db) throws DbException {
                return new TableDataCursor(new BtreeDataTable(btree, tableName, write), db);
            }
        });
    }

    public ICursor lookup(final String indexName, final Object... key) throws DbException {
        final Object[] k = DbUtility.adjustNumberTypes(key);
        return (ICursor) db.runWithLock(new RunnableWithLock() {
            public Object runWithLock(Database db) throws DbException {
                final BtreeDataTable table = new BtreeDataTable(btree, tableName, write);
                checkIndexName(indexName, table);
                return new IndexScopeCursor(table, db, indexName, k, k);
            }
        });
    }

    private Object runWriteTransaction(final ITableRun op) throws DbException {
        return db.runWriteTransaction(new ITransaction() {
            public Object run(Database db) throws DbException {
                final IBtreeDataTable table = new BtreeDataTable(btree, tableName, write);
                try {
                    return op.run(table);
                } finally {
                    table.close();
                }
            }
        });
    }

    public long insert(final Object... values) throws DbException {
        return insertOr(null, values);
    }

    public long insertByFieldNames(final Map<String, Object> values) throws DbException {
        return insertByFieldNamesOr(null, values);
    }

    public long insertWithRowId(final long rowId, final Object... values) throws DbException {
        return insertWithRowIdOr(null, rowId, values);
    }

    public long insertOr(final ConflictAction onConflict, final Object... values) throws DbException {
        return (Long) runWriteTransaction(new ITableRun() {
            public Object run(IBtreeDataTable table) throws DbException {
                return table.insert(onConflict, values);
            }
        });
    }

    public long insertByFieldNamesOr(final ConflictAction onConflict, final Map<String, Object> values)
            throws DbException {
        return (Long) runWriteTransaction(new ITableRun() {
            public Object run(IBtreeDataTable table) throws DbException {
                return table.insert(onConflict, values);
            }
        });
    }

    public long insertWithRowIdOr(final ConflictAction onConflict, final long rowId, final Object... values)
            throws DbException {
        return (Long) runWriteTransaction(new ITableRun() {
            public Object run(IBtreeDataTable table) throws DbException {
                return table.insertWithRowId(onConflict, rowId, values);
            }
        });
    }

    /*
     * (non-Javadoc)
     * 
     * @see org.tmatesoft.rabbit.core.table.ITable#order(java.lang.String)
     */
    public ICursor order(final String indexName) throws DbException {
        return (ICursor) db.runWithLock(new RunnableWithLock() {
            public Object runWithLock(Database db) throws DbException {
                final BtreeDataTable table = new BtreeDataTable(btree, tableName, write);
                checkIndexName(indexName, table);
                return new IndexOrderCursor(table, db, indexName);
            }
        });
    }

    /*
     * (non-Javadoc)
     * 
     * @see org.tmatesoft.rabbit.core.table.ITable#scope(java.lang.String,
     * java.lang.Object[], java.lang.Object[])
     */
    public ICursor scope(final String indexName, final Object[] firstKey, final Object[] lastKey)
            throws DbException {
        final Object[] first = DbUtility.adjustNumberTypes(firstKey);
        final Object[] last = DbUtility.adjustNumberTypes(lastKey);
        return (ICursor) db.runWithLock(new RunnableWithLock() {
            public Object runWithLock(Database db) throws DbException {
                final BtreeDataTable table = new BtreeDataTable(btree, tableName, write);
                checkIndexName(indexName, table);
                if (isNeedReverse(getIndexTable(indexName, table), first, last)) {
                    return new ReverseOrderCursor(new IndexScopeCursor(table, db, indexName, last, first));
                } else {
                    return new IndexScopeCursor(table, db, indexName, first, last);
                }
            }
        });
    }

    public void clear() throws DbException {
        runWriteTransaction(new ITableRun() {
            public Object run(IBtreeDataTable table) throws DbException {
                table.clear();
                return null;
            }
        });
    }

    /**
     * @param indexName
     * @param firstKey
     * @param lastKey
     * @param reverse
     * @param table
     * @return
     * @throws DbException
     */
    private boolean isNeedReverse(final IBtreeIndexTable indexTable, final Object[] firstKey,
            final Object[] lastKey) throws DbException {
        if (firstKey != null && lastKey != null && firstKey.length > 0 && lastKey.length > 0) {
            if (indexTable != null) {
                return indexTable.compareKeys(firstKey, lastKey) < 0;
            } else if (firstKey.length == 1 && lastKey.length == 1 && firstKey[0] instanceof Long
                    && lastKey[0] instanceof Long) {
                return ((Long) firstKey[0]).compareTo((Long) lastKey[0]) > 0;
            }
        }
        return false;
    }

    private IBtreeIndexTable getIndexTable(final String indexName, final BtreeDataTable table) {
        final String index = indexName == null ? table.getPrimaryKeyIndex() : indexName;
        return index != null ? table.getIndex(index) : null;
    }

    private void checkIndexName(final String indexName, final BtreeDataTable table) throws DbException {
        if (!isIndexNameValid(indexName, table)) {
            throw new DbException(DbErrorCode.MISUSE, String.format("Index not exists: %s", indexName));
        }
    }

    private boolean isIndexNameValid(final String indexName, final BtreeDataTable table) {
        if (indexName != null) {
            return getIndexTable(indexName, table) != null;
        } else {
            if (table.getDefinition().isRowIdPrimaryKey()) {
                return true;
            } else {
                return table.getPrimaryKeyIndex() != null;
            }
        }
    }

}
